package ils.infrastructure.messaging;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: Rejeev Divakaran
 * Date: 7/27/13
 */
public class SimpleInMemoryJms {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleInMemoryJms.class);
    private static final Map<String, List<Object>> queueMap = new ConcurrentHashMap<>();
    private static final Map<String, List<Object>> topicMap = new ConcurrentHashMap<>();
    private static final Map<String, List<MessageListener>> queueListenerMap = new ConcurrentHashMap<>();
    private static final Map<String, List<MessageListener>> topicListenerMap = new ConcurrentHashMap<>();

    static {
        new Thread(new MessageDistributor(true)).start();
        new Thread(new MessageDistributor(false)).start();
    }

    public static void send(Object message, String queueName){
        List<Object> messages = queueMap.get(queueName);
        if(messages == null){
            synchronized (queueMap){
                messages = queueMap.get(queueName);
                if(messages == null){
                    messages = new ArrayList<>();
                    queueMap.put(queueName, messages);
                }
            }
        }
        synchronized (messages){
            messages.add(message);
        }
        synchronized (queueMap){
            queueMap.notifyAll();
        }
    }

    public static void publish(Object message, String topicName){
        List<Object> messages = topicMap.get(topicName);
        if(messages == null){
            synchronized (topicMap){
                messages = topicMap.get(topicName);
                if(messages == null){
                    messages = new ArrayList<>();
                    topicMap.put(topicName, messages);
                }
            }
        }
        synchronized (messages){
            messages.add(message);
        }
        synchronized (topicMap){
            topicMap.notifyAll();
        }
    }

    public static void registerQueueListener(MessageListener listener, String queueName){
        List<MessageListener> messageListeners = queueListenerMap.get(queueName);
        if(messageListeners == null){
            synchronized (queueListenerMap){
                messageListeners = queueListenerMap.get(queueName);
                if(messageListeners == null){
                    messageListeners = new ArrayList<>();
                    queueListenerMap.put(queueName, messageListeners);
                }
            }
        }
        synchronized (messageListeners){
            messageListeners.add(listener);
        }
    }

    public static void registerTopicListener(MessageListener listener, String topicName){
        List<MessageListener> messageListeners = topicListenerMap.get(topicName);
        if(messageListeners == null){
            synchronized (topicListenerMap){
                messageListeners = topicListenerMap.get(topicName);
                if(messageListeners == null){
                    messageListeners = new ArrayList<>();
                    topicListenerMap.put(topicName, messageListeners);
                }
            }
        }
        synchronized (messageListeners){
            messageListeners.add(listener);
        }
    }

    private static class MessageDistributor implements Runnable{
        private static final Logger LOGGER = LoggerFactory.getLogger(MessageDistributor.class);
        private boolean isQueue;
        private static final Random random = new Random();

        public MessageDistributor(boolean isQueue){
            this.isQueue = isQueue;
        }

        @Override
        public void run() {
            while (true){
                boolean isEmpty = true;
                for(Map.Entry<String, List<Object>> entry : getMessageMap().entrySet()){
                    List<Object> messageList = entry.getValue();
                    String destinationName = entry.getKey();
                    if(!messageList.isEmpty()){
                        isEmpty = false;
                        Object message = null;
                        synchronized (messageList){
                            if(messageList.isEmpty())continue;
                            message = messageList.remove(0);
                        }
                        distribute(message, destinationName);
                    }
                }
                if(isEmpty){
                    waitOnMessageMap();
                }
            }
        }
        private Map<String, List<Object>> getMessageMap(){
            if(isQueue) return queueMap;
            else return topicMap;
        }
        private void waitOnMessageMap(){
            LOGGER.debug("No messages to distribute, Going to sleep for 30 seconds");
            if(isQueue){
                synchronized (queueMap){
                    try {
                        queueMap.wait(30000);
                    } catch (InterruptedException e) {
                        LOGGER.error("Interrupted", e);
                    }
                }
            }else {
                synchronized (topicMap){
                    try{
                        topicMap.wait(30000);
                    }catch (InterruptedException e){
                        LOGGER.error("Interrupted", e);
                    }
                }
            }
        }

        private void distribute(Object message, String destinationName){
            if(isQueue) handoverQueueMessages(message, destinationName);
            else distributeTopicMessages(message, destinationName);
        }

        private void distributeTopicMessages(Object message, String destinationName) {
            List<MessageListener> messageListeners = topicListenerMap.get(destinationName);
            if(messageListeners == null) return;
            synchronized (messageListeners){
                messageListeners = new ArrayList<>(messageListeners);
            }
            for(MessageListener messageListener : messageListeners){
                try {
                    messageListener.onMessage(message);
                } catch (Exception e) {
                    LOGGER.error("Error while callback for the topic message: " + message, e);
                }
            }
        }

        private void handoverQueueMessages(Object message, String queueName){
            List<MessageListener> messageListeners = queueListenerMap.get(queueName);
            if(messageListeners == null) return;
            MessageListener messageListener = null;
            synchronized (messageListeners){
                int index = random.nextInt(messageListeners.size());
                messageListener = messageListeners.get(index);
            }
            try{
                messageListener.onMessage(message);
            }catch (Exception e){
                LOGGER.error("Error while callback for the queue message: " + message, e);
            }
        }
    }
}
